package drds.binlog.client.adapter.loader;

import drds.binlog.client.adapter.OuterAdapter;
import drds.binlog.protocol.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

/**
 * 适配器工作线程抽象类
 */
public abstract class AbstractAdapterWorker
{

    protected final Logger logger = LoggerFactory.getLogger(this.getClass());

    protected String destination;                                                 // canal实例
    protected List<List<OuterAdapter>> outerAdapterListList;                                               // 外部适配器
    protected ExecutorService groupInnerExecutorService;                                        // 组内工作线程池
    protected volatile boolean running = false;                                                  // 是否运行中
    protected Thread thread = null;
    protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler()
    {

        @Override
        public void uncaughtException(Thread thread, Throwable e)
        {
            logger.error("parse events has an error", e);
        }
    };

    protected void writeOut(final Message message)
    {
        List<Future<Boolean>> futureList = new ArrayList<>();
        // 组间适配器并行运行
        for (List<OuterAdapter> outerAdapterList : outerAdapterListList)
        {
            final List<OuterAdapter> adapters = outerAdapterList;
            futureList.add(groupInnerExecutorService.submit(new Callable<Boolean>()
            {

                @Override
                public Boolean call()
                {
                    try
                    {
                        // 组内适配器穿行运行，尽量不要配置组内适配器
                        for (OuterAdapter outerAdapter : adapters)
                        {
                            long begin = System.currentTimeMillis();
                            outerAdapter.writeOut(message);
                            if (logger.isDebugEnabled())
                            {
                                logger.debug("{} elapsed time: {}",
                                        outerAdapter.getClass().getName(),
                                        (System.currentTimeMillis() - begin));
                            }
                        }
                        return true;
                    } catch (Exception e)
                    {
                        return false;
                    }
                }
            }));

            // 等待所有适配器写入完成
            // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
            for (Future<Boolean> future : futureList)
            {
                try
                {
                    if (!future.get())
                    {
                        logger.error("Outer adapter write failed");
                    }
                } catch (InterruptedException | ExecutionException e)
                {
                    logger.error(e.getMessage(), e);
                }
            }
        }
    }

    public abstract void start();

    public abstract void stop();
}
